package disruptor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import disruptor.event.ValueEvent;
import disruptor.event.handler.ValueEventHandler;
import disruptor.factory.ValueEventFactory;

import java.util.concurrent.Executors;

/**
 * @ClassName MultipleProducerMultipleConsumer
 * @PackageName disruptor
 * @projectName myThread
 * @Description TODO
 * @Author games
 * @Date 2025/9/8 下午5:12
 * @Version 1.0
 */
public class MultipleProducerMultipleConsumer {
    public static void run() throws InterruptedException {
        // 创建 Disruptor - 多生产者模式（修复后写法）
        Disruptor<ValueEvent> disruptor = new Disruptor<>(
                new ValueEventFactory(),
                1024,
                Executors.defaultThreadFactory(),
                com.lmax.disruptor.dsl.ProducerType.MULTI,
                new com.lmax.disruptor.SleepingWaitStrategy()
        );


        // 连接多个事件处理器（每个事件都会被所有处理器处理）
        disruptor.handleEventsWith(
                new ValueEventHandler("Consumer-1"),
                new ValueEventHandler("Consumer-2")
        );

        // 启动 Disruptor
        disruptor.start();

        // 获取 Ring Buffer
        RingBuffer<ValueEvent> ringBuffer = disruptor.getRingBuffer();

        // 创建多个生产者线程
        Thread producer1 = new Thread(() -> {
            for (int i = 0; i < 3; i++) {
                long sequence = ringBuffer.next();
                try {
                    ValueEvent event = ringBuffer.get(sequence);
                    event.setValue(i * 10);
                } finally {
                    ringBuffer.publish(sequence);
                }
            }
        }, "Producer-1");

        Thread producer2 = new Thread(() -> {
            for (int i = 0; i < 3; i++) {
                long sequence = ringBuffer.next();
                try {
                    ValueEvent event = ringBuffer.get(sequence);
                    event.setValue(i * 100);
                } finally {
                    ringBuffer.publish(sequence);
                }
            }
        }, "Producer-2");

        // 启动生产者
        producer1.start();
        producer2.start();

        // 等待生产者完成
        producer1.join();
        producer2.join();

        Thread.sleep(2000); // 等待所有事件处理完成
        disruptor.shutdown();
    }
}

